﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Net;
using System.Threading;
using Microsoft.WindowsAzure;
using Microsoft.WindowsAzure.Diagnostics;
using Microsoft.WindowsAzure.ServiceRuntime;
using Microsoft.WindowsAzure.StorageClient;
using Microsoft.WindowsAzure.Search;
using System.Data.Services.Client;
using System.IO;
using System.Text;

namespace Inverter
{
    public class WorkerRole : RoleEntryPoint
    {
        // client
        private CloudBlobClient blobClient;
        private CloudQueueClient queueClient;

        // blob container
        private CloudBlobContainer container;
        private CloudBlobContainer logContainer;

        // queue
        private CloudQueue inverterQueue;
        private CloudQueue queue;
        private CloudQueue masterQueue;

        // table context
        private IndexContext context;

        private int start;

        private StringBuilder log = new StringBuilder();
    
        private void Initialize()
        {
            CloudStorageAccount indexAccount = CloudStorageAccount.FromConfigurationSetting("IndexConnectionString");

            queueClient = indexAccount.CreateCloudQueueClient();
            blobClient = indexAccount.CreateCloudBlobClient();

            inverterQueue = queueClient.GetQueueReference("inverter");
            inverterQueue.CreateIfNotExist();
            masterQueue = queueClient.GetQueueReference("master");
            masterQueue.CreateIfNotExist();

            // container
            container = blobClient.GetContainerReference("temp");
            logContainer = blobClient.GetContainerReference("log");
            logContainer.CreateIfNotExist();

            // table context
            CloudTableClient tableClient = indexAccount.CreateCloudTableClient();
            tableClient.CreateTableIfNotExist("index");
            context = new IndexContext(indexAccount.TableEndpoint.ToString(), indexAccount.Credentials);
        }

        public override bool OnStart()
        {
            // Set the maximum number of concurrent connections 
            ServicePointManager.DefaultConnectionLimit = 12;

            // For information on handling configuration changes
            // see the MSDN topic at http://go.microsoft.com/fwlink/?LinkId=166357.

            #region Setup CloudStorageAccount Configuration Setting Publisher

            CloudStorageAccount.SetConfigurationSettingPublisher(
                (configName, configSetter) =>
                {
                    // Provide the configSetter with the initial value
                    configSetter(RoleEnvironment.GetConfigurationSettingValue(configName));
                    RoleEnvironment.Changed += (s, arg) =>
                    {
                        if (arg.Changes.OfType<RoleEnvironmentConfigurationSettingChange>()
                            .Any((change) => (change.ConfigurationSettingName == configName)))
                        {
                            if (!configSetter(RoleEnvironment.GetConfigurationSettingValue(
                                configName)))
                                // In this case, the change to the storage account credentials
                                // in the service configuration is significant enough that the
                                // role needs to be recycled in order to use the latest settings.
                                // (for example, the endpoint has changed)
                                RoleEnvironment.RequestRecycle();
                        }
                    };
                });
            #endregion

            Initialize();

            return base.OnStart();
        }

        public override void Run()
        {
            while (true)
            {
                CloudQueueMessage m = inverterQueue.GetMessage();

                if (m != null)
                {
                    Stopwatch watch = new Stopwatch();
                    watch.Start();

                    inverterQueue.DeleteMessage(m);
                    Debug.WriteLine("Start Inverter " + m.AsString);
                    start = Convert.ToInt32(m.AsString);

                    // <start number>
                    CloudBlob blob = container.GetBlobReference(m.AsString);
                    String content = blob.DownloadText();
                    blob.DeleteIfExists();

                    // construct inverted index
                    Invert(content);
                    Debug.WriteLine("**********Invert " + m.AsString + " done**********");

                    // send to master done
                    CloudQueueMessage doneMessage = new CloudQueueMessage(m.AsString);
                    masterQueue.AddMessage(doneMessage);

                    watch.Stop();
                    Debug.WriteLine("Invert Time: " + watch.ElapsedMilliseconds + " ms");
                    log.Append("Invert Time: " + watch.ElapsedMilliseconds + " ms");
                    CloudBlob logBlob = logContainer.GetBlobReference("inverter" + start);
                    logBlob.UploadText(log.ToString());
                }

                Thread.Sleep(2000);
            }
        }

        private void Invert(String content)
        {
            Dictionary<String, List<Posting>> dictionary = new Dictionary<String, List<Posting>>();

            String[] items = Tokenizer.LineTokenize(content);

            // deal with every intermediate document from parsing
            for (int i = 0; i + 1 < items.Length; i = i + 2)
            {
                int docID = Convert.ToInt32(items[i]);

                Dictionary<string, double> termDic = Parse(items[i + 1]);

                foreach (KeyValuePair<string, double> pair in termDic)
                {
                    if (!dictionary.ContainsKey(pair.Key))
                    {
                        List<Posting> posting_list = new List<Posting>();
                        Posting posting = new Posting(docID, pair.Value);
                        posting_list.Add(posting);
                        dictionary.Add(pair.Key, posting_list);
                    }
                    else
                    {
                        // existed term
                        List<Posting> posting_list = dictionary[pair.Key];
                        Posting posting = new Posting(docID, pair.Value);
                        posting_list.Add(posting);
                    }
                }
            }

            
            // write index to table
            int k = 0;
            foreach (KeyValuePair<String, List<Posting>> pair in dictionary)
            {
                String term = pair.Key;

                StringBuilder str = new StringBuilder();
                bool first = true;
                foreach (Posting posting in pair.Value)
                {
                    if (first)
                    {
                        str.Append(posting.DocID + "," + posting.TermFreq);
                        first = false;
                    }
                    else
                        str.Append("," + posting.DocID + "," + posting.TermFreq);
                }

                context.Add(start.ToString(), term, str.ToString(), pair.Value.Count);
                k++;
                if (k == 100)
                {
                    context.SaveChanges(SaveChangesOptions.Batch);
                    k = 0;
                }
            }
            context.SaveChanges(SaveChangesOptions.Batch);
        }

        private Dictionary<string, double> Parse(string content)
        {
            Dictionary<string, double> terms = new Dictionary<string, double>();
            string[] items = Tokenizer.CommaTokenize(content);
            for (int i = 0; i < items.Length; i++)
            {
                terms.Add(items[i], Convert.ToDouble(items[++i]));
            }
            return terms;
        }
    }
}
